package MapJoin;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.*;
import java.net.URI;
import java.util.HashMap;

public class MJMapper extends Mapper<LongWritable, Text,Text, NullWritable> {

    /*
    * 只做一次
    * 完成分布式缓存
    * 01    小米
    * 02    华为
    * 03    格力
    * */

    //初始化操作：加载数据信息
    HashMap<String,String> pdMap= new HashMap();

    /*
     * setup():在执行Map之前进行相关变量或者资源的集中初始化工作，只执行一次
     * 简单的说顶级接口是为了在map或是reduce任务中跟踪task的状态，很自然的MapContext就是记录了map执行的上下文，
     * 在mapper类中，这个context可以存储一些job conf的信息，如的运行时参数等，
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        //获取Driver端缓存文件的路径
        URI[] cacheFiles = context.getCacheFiles();
        String path = cacheFiles[0].getPath();
        //获取分布式缓存文件
        BufferedReader reader = new BufferedReader(
                                          new InputStreamReader(
                                             new FileInputStream(
                                                     //小文件
                                                    new File(path)), "UTF-8"));
       String line ;
       //判断是否为空
       while(!StringUtils.isEmpty(line=reader.readLine())){
          //处理数据
           String[] fileds = line.split("\t");

          //加入缓存数据到集合中
           pdMap.put(fileds[0],fileds[1]);


       }
    }
    //读到的是大文件（order.txt）的数据
           /*201801	01	1
            201802	02	2
            201803	03	3
            201804	01	4
            201805	02	5
            201806	03	6*/

           /*201801	小米	1
            201802	华为	2
            201803	格力	3
            201804	小米	4
            201805	华为	5
            201806	格力	6*/
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
      //获取数据
        String line = value.toString();
       //截取切分数据
        String[]  fields= line.split("\t");
       //获取产品id (key)
        String p_id=fields[1];
        //获取商品名称(value)
        String pd_name = pdMap.get(p_id);
       //重新拼接输出数据
       String newStr=fields[0]+"\t"+pd_name+"\t"+fields[2];
       Text k = new Text();
       k.set(newStr);
       //写出
        context.write(k,NullWritable.get());

    }
}
